package com.flute.haflute.jobbox.child;

import java.util.*;
import org.slf4j.*;

import com.flute.haflute.jobbox.communication.ChildCommunicationService;
import com.flute.haflute.jobbox.communication.messages.*;

/**
 * 任务调度器的管理者
 */
public class TaskSchedulerManager {

	Logger logger = LoggerFactory.getLogger(TaskSchedulerManager.class);

	/** 调度器池，映射为<任务类型,调度器> */
	private Map<String, TaskScheduler> schedulers;

	/** 通信平台服务 */
	private ChildCommunicationService communicationService;

	public TaskSchedulerManager() {
		schedulers = new HashMap<String, TaskScheduler>();
		communicationService = new ChildCommunicationService();
		initSchedulers();
	}

	/** 初始化调度器并放入池子 */
	private void initSchedulers() {
		CloudTaskScheduler cloudScheduler = new CloudTaskScheduler(
				communicationService);
		schedulers.put(CloudJobReqMessage.type, cloudScheduler);

		MRTaskScheduler mrScheduler = new MRTaskScheduler(communicationService);
		schedulers.put(MapperReqMessage.type, mrScheduler);

		FileTransferTask fileReiverTask = new FileTransferTask();
		schedulers.put(FileTransferMessage.type, fileReiverTask);
	}

	/** 准备提交任务给任务调度器处理 */
	public void prepare2runTask(MessageBase request) {
		TaskScheduler scheduler = schedulers.get(request.getType());
		if (scheduler == null) {
			logger.error("no such task runner for request type: {}",
					request.getType());
			return;
		}
		try {
			scheduler.scheduleTask(request);
		} catch (Throwable e) {
			String msg = request.getType();
			logger.error(msg, e);
		}
	}
}
